package com.atguigu.chapter5.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;


import java.util.ArrayList;

/**
 * @ClassName: RedisSinkDemo
 * @Description:
 * @Author: kele
 * @Date: 2021/4/5 22:03
 **/
public class RedisSinkDemo {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);


        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 30));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

        DataStreamSource<WaterSensor> ds = env.fromCollection(waterSensors);

        FlinkJedisPoolConfig redisConf = new FlinkJedisPoolConfig.Builder()
                .setHost("hadoop162")
                .setPort(6379)
                .setMaxTotal(100)
                .setMaxIdle(10)
                .setTimeout(10 * 1000)
                .build();


        ds.keyBy(WaterSensor::getId)
                .sum("vc")
                .addSink(new RedisSink<>(redisConf, new RedisMapper<WaterSensor>() {

                    //获取命令操作
                    @Override
                    public RedisCommandDescription getCommandDescription() {
                        //如果不是hash或者SORTED_SET，第二个参数会被忽略
                        //如果是，则是hash的key值
                        return new RedisCommandDescription(RedisCommand.SET,"");
//                        return new RedisCommandDescription(RedisCommand.HSET,"senion");
                    }

                    //获取key，如果时hash，则是内部key值
                    @Override
                    public String getKeyFromData(WaterSensor data) {
                        return data.getId();
                    }

                    //存储的值
                    @Override
                    public String getValueFromData(WaterSensor data) {
                        return JSON.toJSONString(data);
                    }
                }));


        env.execute();


    }

}
